package com.ucas.kafka.consumer;

import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.ucas.kafka.protobuf.AlarmEntity;
import com.google.protobuf.InvalidProtocolBufferException;
import com.ucas.kafka.protobuf.Lns.EuEntity;
import com.ucas.kafka.protobuf.Lns.ReplayResponse;
import com.ucas.kafka.protobuf.TCPMsgEntity.TcpCmdExecState;

@Component
public class kafkaConsumer {
	//指令回令信息--topic
	@KafkaListener(topics={"vela-tcpmsg"})
	public void velaTcpMsg(ConsumerRecord<String, byte[]> record) {
        try {
        	TcpCmdExecState entiy = TcpCmdExecState.parseFrom(record.value());
        	System.err.println("kafkaConsumer--vela-tcpmsg:");
        	System.out.println(entiy);
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
	}
	//遥测参数观察器--topic
	@KafkaListener(topics={"engbc"})
	public void engbcMsg(ConsumerRecord<String, byte[]> record) {
        try {
        	ReplayResponse entiy = ReplayResponse.parseFrom(record.value());
        	System.err.println("kafkaConsumer--engbc:");
        	List<EuEntity> eulist = entiy.getEuList();
        	for(EuEntity eu:eulist) {
        		System.out.println("eu="+eu.getEu()+"-----Tmid=:"+eu.getTmid());
        	}
//        	System.out.println(entiy);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	//报警监控--topic
	@KafkaListener(topics={"alarm"})
	public void alarm(ConsumerRecord<String, byte[]> record) {
        try {
        	AlarmEntity.Alarm entiy = AlarmEntity.Alarm.parseFrom(record.value());
        	System.err.println("kafkaConsumer--alarm:");
        	System.out.println(entiy);
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
	}
}
